package com.xxl.job.admin.core.thread;

import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.model.XxlJobAdminMachine;
import com.xxl.job.admin.core.model.XxlJobInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.CollectionUtils;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 定时任务分配
 *
 * @author chenlinqing
 * @description:
 * @date: 2025/8/7 18:16
 */
public class JobAllocationHelper {
    private static Logger logger = LoggerFactory.getLogger(JobAllocationHelper.class);

    private static JobAllocationHelper instance = new JobAllocationHelper();

    public static JobAllocationHelper getInstance() {
        return instance;
    }


    private Thread xxlJobAdminMachineRegistryThread;
    private Thread deathMachineJobAllocationThread;
    private volatile boolean xxlJobAdminMachineRegistryThreadToStop = false;
    private volatile boolean deathMachineJobAllocationThreadToStop = false;

    public void start() {

        // registry machine
        xxlJobAdminMachineRegistryThread = new Thread(new Runnable() {
            @Override
            public void run() {
                while (!xxlJobAdminMachineRegistryThreadToStop) {
                    try {
                        long startTime = System.currentTimeMillis();
                        XxlJobAdminMachine machine = new XxlJobAdminMachine();
                        String address = null;
                        address = XxlJobAdminConfig.getAdminConfig().getMachineAddress();
                        machine.setAddress(address);
                        machine.setUpdateTime(new Date());
                        XxlJobAdminConfig.getAdminConfig().getXxlJobAdminMachineDao().insertOrUpdate(machine);
                        long processTimeCost = System.currentTimeMillis() - startTime;
                        TimeUnit.MILLISECONDS.sleep(XxlJobAdminConfig.getAdminConfig().getMachineRefreshTime() * 1000 - processTimeCost);
                    } catch (Exception e) {
                        if (!xxlJobAdminMachineRegistryThreadToStop) {
                            logger.error(">>>>>>>>>>> xxl-job, JobAllocationHelper#xxlJobAdminMachineRegistryThread error:{}", e);
                        }
                    }
                }
            }
        });

        deathMachineJobAllocationThread = new Thread(new Runnable() {
            @Override
            public void run() {

                while (!deathMachineJobAllocationThreadToStop) {
                    Connection conn = null;
                    Boolean connAutoCommit = null;
                    PreparedStatement preparedStatement = null;
                    try {

                        conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
                        connAutoCommit = conn.getAutoCommit();
                        conn.setAutoCommit(false);

                        //数据库锁，保证线程安全
                        preparedStatement = conn.prepareStatement("select * from xxl_job_lock where lock_name = 'job_allocation' for update");
                        preparedStatement.execute();

                        List<XxlJobAdminMachine> survivalMachineList = XxlJobAdminConfig.getAdminConfig()
                                .getXxlJobAdminMachineDao()
                                .getSurvivalMachineList(XxlJobAdminConfig.getAdminConfig().getMachineDeathTime());
                        if (!CollectionUtils.isEmpty(survivalMachineList)) {
                            List<String> survivalMachineAddressList = survivalMachineList.stream().map(XxlJobAdminMachine::getAddress).toList();
                            List<XxlJobInfo> notAllocationJobs = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao()
                                    .getJobsNotAllocationToSurvivalMachine(survivalMachineAddressList);

                            if (!CollectionUtils.isEmpty(notAllocationJobs)) {
                                for (int i = 0; i < notAllocationJobs.size(); i++) {
                                    XxlJobInfo jobInfo = notAllocationJobs.get(i);
                                    jobInfo.setAdminMachineAddress(survivalMachineList.get(i % survivalMachineList.size()).getAddress());
                                    XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().update(jobInfo);
                                }
                            }
                        }
                    } catch (Exception e) {
                        if (!deathMachineJobAllocationThreadToStop) {
                            logger.error(">>>>>>>>>>> xxl-job, JobAllocationHelper#deathMachineJobAllocationThread error:{}", e);
                        }
                    } finally {
                        // commit
                        if (conn != null) {
                            try {
                                conn.commit();
                            } catch (Throwable e) {
                                logger.error(e.getMessage(), e);
                            }
                            try {
                                conn.setAutoCommit(connAutoCommit);
                            } catch (Throwable e) {
                                logger.error(e.getMessage(), e);
                            }
                            try {
                                conn.close();
                            } catch (Throwable e) {
                                logger.error(e.getMessage(), e);
                            }
                        }

                        // close PreparedStatement
                        if (null != preparedStatement) {
                            try {
                                preparedStatement.close();
                            } catch (Throwable e) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                    }

                }
                logger.info(">>>>>>>>>>> xxl-job, JobAllocationHelper#deathMachineJobAllocationThread  stop");

            }
        });

        xxlJobAdminMachineRegistryThread.setDaemon(true);
        xxlJobAdminMachineRegistryThread.setName("xxl-job, admin JobAllocationHelper#xxlJobAdminMachineRegistryThread");
        xxlJobAdminMachineRegistryThread.start();

        deathMachineJobAllocationThread.setDaemon(true);
        deathMachineJobAllocationThread.setName("xxl-job, admin JobAllocationHelper#deathMachineJobAllocationThread");
        deathMachineJobAllocationThread.start();
    }

    public void toStop() {
        xxlJobAdminMachineRegistryThreadToStop = true;
        deathMachineJobAllocationThreadToStop = true;
        // interrupt and wait
        xxlJobAdminMachineRegistryThread.interrupt();
        deathMachineJobAllocationThread.interrupt();
        try {
            xxlJobAdminMachineRegistryThread.join();
            deathMachineJobAllocationThread.join();
        } catch (Throwable e) {
            logger.error(e.getMessage(), e);
        }
    }

}
